package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.MinStockWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.map.StockPutHdfsMap;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.fs.StringWriter;
import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink;
import org.apache.flink.streaming.connectors.fs.bucketing.DateTimeBucketer;

/**
 * 分时行情数据备份至HDFS
 */
public class StockMinHdfsTask implements ProcessDataInterface {
    /**
     * 开发步骤：
     *      1.设置HDFS存储路径
     *      2.设置数据文件参数
     *        （大小、分区、格式、前缀、后缀）
     *      3.数据分组
     *      4.划分时间窗口
     *      5.数据处理
     *      6.转换并封装数据
     *      7.写入HDFS
     */


    @Override
    public void process(DataStream<CleanBean> waterData) {
        //todo 1.设置HDFS存储路径
        //获取写入对象（1.7.0版本的api，其他的还有StreamingFileSink）
        BucketingSink<String> bucketingSink = new BucketingSink<>(QuotConfig.config.getProperty("stock.sec.hdfs.path"));
        //todo 2.设置数据文件参数
        //todo   （大小、分区、格式、前缀、后缀）
        bucketingSink.setBatchSize(Long.parseLong(QuotConfig.config.getProperty("hdfs.batch")));
        //按时间分区，会自动识别
        bucketingSink.setBucketer(new DateTimeBucketer<>(QuotConfig.config.getProperty("hdfs.bucketer")));
        bucketingSink.setWriter(new StringWriter<>());
        bucketingSink.setInProgressPrefix("stock-");//正在写入的文件
        bucketingSink.setPendingPrefix("stock2-"); //挂起的文件
        bucketingSink.setInProgressSuffix(".txt");
        bucketingSink.setPendingSuffix(".txt");
        //todo 3.数据分组
        waterData.keyBy(new KeyFunction())
                //todo 4.划分时间窗口
                .timeWindow(Time.minutes(1))
                //todo 5.数据处理
                .apply(new MinStockWindowFunction())
                //todo 6.转换并封装数据
                .map(new StockPutHdfsMap())
                //todo 7.写入HDFS
                .addSink(bucketingSink);
    }
}
